package com.yww.datastructure.blockingqueue;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author yww
 * @description 双锁实现阻塞队列
 * @since 2024/2/7 14:15
 */
@SuppressWarnings("unchecked")
public final class BlockingQueue2<E> implements BlockingQueue<E> {
    private final E[] array;
    private final AtomicInteger size = new AtomicInteger(0);  // 使用原子变量，对size进行原子操作
    private int head;
    private int tail;

    private final Lock headLock = new ReentrantLock();  // 消费锁
    private final Lock tailLock = new ReentrantLock(); // 生产锁
    private final Condition headWaits = headLock.newCondition(); // 消费等待队列
    private final Condition tailWaits = tailLock.newCondition(); // 生产等待队列

    public BlockingQueue2(int capacity) {
        this.array = (E[]) new Object[capacity];
    }

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly(); // 获取生产锁
        int c;
        try {
            while (isFull()) {
                tailWaits.await(); // 队列满，生产线程等待
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement(); // size 自增原子操作

            // headWaits.signal()  唤醒一个消费线程? no,非嵌套加锁，避免死锁

            if (c + 1 < array.length) {
                tailWaits.signal();  // 自增没有达到容量上限，串行唤醒生产线程
            }
        } finally {
            tailLock.unlock();
        }

        if (c == 0) { // 非嵌套加锁，避免死锁，如果队列由空转换为不空，size:0->1，唤醒一个消费线程
            // 获取消费线程锁
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;
        try {
            long n = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()) {
                if (n <= 0) { // 超时
                    return false;
                }
                n = tailWaits.awaitNanos(n);
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            if (c + 1 < array.length) {
                tailWaits.signal(); // 说明从不满 -> 不满，串行唤醒其他生产线程
            }
        } finally {
            tailLock.unlock();
        }
        if (c == 0) {
            headLock.lock();
            try {
                headWaits.signal(); // 从空 -> 不空，唤醒一个消费线程
            } finally {
                headLock.unlock();
            }
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        headLock.lockInterruptibly();
        E returnVal;
        int c;
        try {
            while (isEmpty()) {
                headWaits.await();
            }
            returnVal = array[head];
            array[head] = null; // help gc
            if (++head == array.length) {
                head = 0;
            }
            c = size.getAndDecrement(); // 原子自减
            if (c > 1) {
                headWaits.signal(); // 如果消费完成还至少有一个元素，串行唤醒消费线程
            }
        } finally {
            headLock.unlock();
        }
        if (c == array.length) { // 队列从满到不满，唤醒一个生产线程
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return returnVal;
    }

    @Override
    public E poll(long timeout) throws InterruptedException {
        headLock.lockInterruptibly();
        E returnVal;
        int c;
        try {
            long n = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isEmpty()) {
                if (n <= 0) { // 等待超时，返回nully
                    return null;
                }
                n = headWaits.awaitNanos(n);
            }
            returnVal = array[head];
            array[head] = null; // help gc
            if (++head == array.length) {
                head = 0;
            }
            c = size.getAndDecrement(); // 原子自减
            if (c > 1) {
                headWaits.signal(); // 如果消费完成还至少有一个元素，串行唤醒消费线程
            }
        } finally {
            headLock.unlock();
        }
        if (c == array.length) { // 队列从满到不满，唤醒一个生产线程
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return returnVal;
    }

    @Override
    public boolean isEmpty() {
        return size.get() == 0;
    }

    @Override
    public boolean isFull() {
        return size.get() == array.length;
    }

    @Override
    public int size() {
        return size.get();
    }
}
